package com.dl.base.api.websocket

import android.os.Looper
import android.os.SystemClock
import com.dl.base.AppContext.url
import com.dl.base.api.websocket.bean.WebSocketInfo
import com.dl.base.api.websocket.cache.WebSocketInfoCachePool
import com.dl.base.api.websocket.constant.WebSocketCloseEnum
import com.dl.base.api.websocket.exception.ImproperCloseException
import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.core.ObservableEmitter
import io.reactivex.rxjava3.core.ObservableOnSubscribe
import io.reactivex.rxjava3.functions.Predicate
import okhttp3.*
import okio.ByteString
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
import kotlin.Function as Function1001

class WebSocketWorkerImpl(
    val client: OkHttpClient,
    private val reconnectInterval: Long,//重连间隔时间
    private val reconnectIntervalTimeUnit: TimeUnit//重连间隔时间的单位
) {
    /**
     * 缓存观察者对象，Url对应一个Observable
     */
    private var observableCacheMap: ConcurrentHashMap<String, Observable<WebSocketInfo>> =
        ConcurrentHashMap(16)

    /**
     * 缓存Url和对应的WebSocket实例，同一个Url共享一个WebSocket连接
     */
    private var webSocketPool: ConcurrentHashMap<String, WebSocket> = ConcurrentHashMap(16)

    /**
     * WebSocketInfo缓存池
     */
    private var webSocketInfoPool: WebSocketInfoCachePool = WebSocketInfoCachePool(20)

    operator fun get(url: String): Observable<WebSocketInfo> {
        return getWebSocketInfo(url)
    }

    operator fun get(url: String, timeout: Long, timeUnit: TimeUnit): Observable<WebSocketInfo> {
        return getWebSocketInfo(url, timeout, timeUnit)
    }

    fun send(url: String, msg: String): Observable<Boolean> {
        return Observable.create { emitter ->
            val webSocket = webSocketPool[url]
            if (webSocket == null) {
                emitter.onError(IllegalStateException("The WebSocket not open"))
            } else {
                emitter.onNext(webSocket.send(msg))
            }
        }
    }

    fun send(url: String, byteString: ByteString): Observable<Boolean> {
        return Observable.create { emitter ->
            val webSocket = webSocketPool[url]
            if (webSocket == null) {
                emitter.onError(IllegalStateException("The WebSocket not open"))
            } else {
                emitter.onNext(webSocket.send(byteString))
            }
        }
    }

    fun asyncSend(url: String, msg: String): Observable<Boolean> {
        return getWebSocket(url)
            .take(1)
            .map { webSocket -> webSocket.send(msg) }
    }

    fun asyncSend(url: String, byteString: ByteString): Observable<Boolean> {
        return getWebSocket(url)
            .take(1)
            .map { webSocket -> webSocket.send(byteString) }
    }

    fun close(url: String): Observable<Boolean> {
        return Observable.create<WebSocket> { emitter ->
            val webSocket = webSocketPool[url]
            if (webSocket == null) {
                emitter.onError(NullPointerException("url:$url WebSocket must be not null"))
            } else {
                emitter.onNext(webSocket)
            }
        }.map { webSocket -> closeWebSocket(webSocket) }
    }

    fun closeNow(url: String): Boolean {
        observableCacheMap.remove(url)
        return closeWebSocket(webSocketPool[url])
    }

    fun closeAllNow() {
        for ((_, value) in webSocketPool) {
            closeWebSocket(value)
        }
    }

    /**
     * 是否有连接
     */
    private fun hasWebSocketConnection(url: String): Boolean {
        return webSocketPool[url] != null
    }

    private var isCanceling = false

    /**
     * 关闭WebSocket连接
     */
    private fun closeWebSocket(webSocket: WebSocket?): Boolean {
        if (webSocket == null) {
            return false
        }
        /*
            使用close无法关闭，使用cancel会触发onFailure回调
            因为onFailure中需要维护重连的逻辑，所以使用isCanceling字段
            isCanceling为true时触发onFailure不会调用重连逻辑
             */isCanceling = true
        val normalCloseEnum = WebSocketCloseEnum.USER_EXIT
        val result = webSocket.close(normalCloseEnum.code, normalCloseEnum.reason)
        webSocket.cancel()
        if (result) {
            removeUrlWebSocketMapping(webSocket)
        }
        return result
    }

    /**
     * 移除Url和WebSocket的映射
     */
    private fun removeUrlWebSocketMapping(webSocket: WebSocket) {
        val iterator = webSocketPool.iterator()
        while (iterator.hasNext()) {
            val next = iterator.next()
            if (next.value === webSocket) {
                observableCacheMap.remove(url)
                iterator.remove()
            }
        }
    }

    private fun removeWebSocketCache(webSocket: WebSocket) {
        for ((url, value) in webSocketPool) {
            if (value === webSocket) {
                webSocketPool.remove(url)
            }
        }
    }

    private fun getWebSocket(url: String): Observable<WebSocket> {
        return getWebSocketInfo(url).filter(Predicate { webSocketInfo -> webSocketInfo.webSocket != null })
            .map { webSocketInfo -> webSocketInfo.webSocket }
    }

    private fun getWebSocketInfo(url: String): Observable<WebSocketInfo> {
        return getWebSocketInfo(url, 5, TimeUnit.SECONDS)
    }

    @Synchronized
    fun getWebSocketInfo(
        url: String,
        timeout: Long,
        timeUnit: TimeUnit
    ): Observable<WebSocketInfo> {
        //先从缓存中取
        var observable = observableCacheMap[url]
        if (observable == null) {
            //缓存中没有，新建
            observable = Observable
                .create(WebSocketOnSubscribe(url))
                .retry() //因为有share操作符，只有当所有观察者取消注册时，这里才会回调
                .doOnDispose { //所有都不注册了，关闭连接
                    closeNow(url)
                } //Share操作符，实现多个观察者对应一个数据源
                .share()
            //将数据源缓存
            observableCacheMap[url] = observable
        } else {
            //缓存中有，从连接池中取出
            val webSocket = webSocketPool[url]
            if (webSocket != null) {
                observable = observable.startWithItem(createConnect(url, webSocket))
            }
        }
        return observable!!
    }


    private var timeoutStart = 0L

    /**
     * 组装数据源
     */
    private inner class WebSocketOnSubscribe(private val webSocketUrl: String) :
        ObservableOnSubscribe<WebSocketInfo> {
        private var webSocket: WebSocket? = null
        private var isReconnecting = false

        @Throws(Exception::class)
        override fun subscribe(emitter: ObservableEmitter<WebSocketInfo>) {
            //因为retry重连不能设置延时，所以只能这里延时，降低发送频率
            if (webSocket == null && isReconnecting) {
                if (Thread.currentThread() !== Looper.getMainLooper().thread) {
                    val timeout: Long = reconnectIntervalTimeUnit.toMillis(reconnectInterval)
                    if (timeoutStart == 0L) {
                        timeoutStart = System.currentTimeMillis()
                    }
                    if (System.currentTimeMillis() - timeoutStart < timeout) {
                        SystemClock.sleep(500)
                        initWebSocket(emitter)
                    } else {
                        timeoutStart = 0L
                        if (!emitter.isDisposed) {
                            emitter.onNext(createConnectTimeOut(webSocketUrl))
                        }
                        closeNow(webSocketUrl)
                        isCanceling = false
                    }
                }
            } else {
                initWebSocket(emitter)
            }
        }

        private fun createRequest(url: String): Request {
            return Request.Builder().get().url(url).build()
        }

        /**
         * 初始化WebSocket
         */
        @Synchronized
        private fun initWebSocket(emitter: ObservableEmitter<WebSocketInfo>) {
            if (webSocket == null) {
                webSocket = client.newWebSocket(
                    createRequest(webSocketUrl),
                    object : WebSocketListener() {
                        override fun onOpen(webSocket: WebSocket, response: Response) {
                            super.onOpen(webSocket, response)
                            //重置重新连接时间
                            timeoutStart = 0L
                            //连接成功
                            if (!emitter.isDisposed) {
                                webSocketPool[webSocketUrl] = this@WebSocketOnSubscribe.webSocket!!
                                //重连成功
                                if (isReconnecting) {
                                    emitter.onNext(createReconnect(webSocketUrl, webSocket))
                                } else {
                                    emitter.onNext(createConnect(webSocketUrl, webSocket))
                                }
                            }
                            isReconnecting = false
                        }

                        private var time = 0L
                        override fun onMessage(webSocket: WebSocket, text: String) {
                            super.onMessage(webSocket, text)
//                            Log.i("aaaaa", text)
//                            val currentTimeMillis = System.currentTimeMillis()
//                            Log.i("aaaaa", "" + (currentTimeMillis - time))
//                            time = currentTimeMillis
                            //收到消息
                            if (!emitter.isDisposed) {
                                emitter.onNext(
                                    createReceiveStringMsg(
                                        webSocketUrl,
                                        webSocket,
                                        text
                                    )
                                )
                            }
                        }

                        override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
                            super.onMessage(webSocket, bytes)
                            //收到消息
                            if (!emitter.isDisposed) {
                                emitter.onNext(
                                    createReceiveByteStringMsg(
                                        webSocketUrl,
                                        webSocket,
                                        bytes
                                    )
                                )
                            }
                        }

                        override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
                            super.onClosed(webSocket, code, reason)
                            if (!emitter.isDisposed) {
                                emitter.onNext(createClose(webSocketUrl))
                            }
                        }

                        override fun onFailure(
                            webSocket: WebSocket,
                            t: Throwable,
                            response: Response?
                        ) {
                            super.onFailure(webSocket, t, response)
                            if (!isCanceling) {
                                isReconnecting = true
                                this@WebSocketOnSubscribe.webSocket = null
                                if (!emitter.isDisposed) {
                                    emitter.onNext(createPrepareReconnect(webSocketUrl))
                                    //失败发送onError，让retry操作符重试
                                    emitter.onError(ImproperCloseException())
                                }
                                //移除WebSocket缓存，retry重试重新连接
                                removeWebSocketCache(webSocket)
                            } else {
                                if (!emitter.isDisposed) {
                                    emitter.onNext(createClose(webSocketUrl))
                                }
                                isCanceling = false
                            }
                        }
                    })
            }
        }
    }

    private fun createConnect(url: String, webSocket: WebSocket): WebSocketInfo {
        return webSocketInfoPool.obtain(url).apply {
            this.webSocket = webSocket
            this.isConnect = true
        }
    }

    private fun createReconnect(url: String, webSocket: WebSocket): WebSocketInfo {
        return webSocketInfoPool.obtain(url).apply {
            this.webSocket = webSocket
            this.isConnect = true
            this.isReconnect = true
        }
    }

    private fun createConnectTimeOut(url: String): WebSocketInfo {
        return webSocketInfoPool.obtain(url).apply {
            this.isConnectTimeOut = true
        }
    }

    private fun createPrepareReconnect(url: String): WebSocketInfo {
        return webSocketInfoPool.obtain(url).apply {
            this.isPrepareReconnect = true
        }
    }

    private fun createReceiveStringMsg(
        url: String,
        webSocket: WebSocket,
        stringMsg: String
    ): WebSocketInfo {
        return webSocketInfoPool.obtain(url).apply {
            this.isConnect = true
            this.webSocket = webSocket
            this.stringMsg = stringMsg
        }
    }

    private fun createReceiveByteStringMsg(
        url: String,
        webSocket: WebSocket,
        byteMsg: ByteString
    ): WebSocketInfo {
        return webSocketInfoPool.obtain(url).apply {
            this.isConnect = true
            this.webSocket = webSocket
            this.byteStringMsg = byteMsg
        }
    }

    private fun createClose(url: String): WebSocketInfo {
        return webSocketInfoPool.obtain(url)
    }
}